package com.hr.reactive.hotAndcold;

import com.hr.reactive.common.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;

import java.time.Duration;

/**
 * @author Administrator
 */
public class Lec04ReplayCache {
    private static final Logger log = LoggerFactory.getLogger(Lec04ReplayCache.class);

    public static void main(String[] args) {
        //autoConnect 没有订阅者还在生产数据
        var movie = stockStream().replay(2).autoConnect(0);
        Utils.sleepSeconds(2000);
        log.info("sam joining");
        movie.take(4).subscribe(Utils.subscriber("sam"));
        Utils.sleepSeconds(5000);
        log.info("mike joining");
        movie.take(4).subscribe(Utils.subscriber("mike"));
        Utils.sleepSeconds(10000);
    }

    private static Flux<Integer> stockStream() {
        return Flux.generate(sink -> sink.next(Utils.faker().random().nextInt(10, 100)))

                .delayElements(Duration.ofSeconds(1))
                .doOnNext(price -> log.info("emitting price:{}", price))
                .cast(Integer.class);

    }
}
